package com.hjl.demo.kafka;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/*
业务数据，从mysql读取数据，写入Kafka
ods层
 */
public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);

    /*    //1.1设置状态后端FS并且开启CK  memory FS rocksdb （为了安全）
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020//gmall-flink/ck"));

        env.enableCheckpointing(5000L);//没五秒触发一次 头头
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);//头尾*/
        //2.通过FlinkDCD构建SourceFunction并读取数据
        //MySQL的builder方法 返回的是泛型 再配置mysql的参数
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .serverTimeZone("Asia/Shanghai")
                .hostname("localhost")
                .port(3307)
                .username("root")
                .password("chickenkang.")
                .databaseList("practice")//读取哪个数据库
                //.tableList()//哪个表
                .deserializer(new CustomerDeserialization())//官方告诉需要反序列化
                .startupOptions(StartupOptions.latest())
                .build();
        //将源分配进来，得到流
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        //3.打印数据并写入Kafka
        streamSource.print();
        String sinkTopic = "first";
        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //4.启动任务
        env.execute("FlinkCDCWithCustomerDeserialization");
    }
}
